package io.zhengsh.juc._5disruptor;

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.Data;

import java.util.concurrent.Executors;

public class DisruptorTest {

    // 5. 测试代码
    public static void main(String[] args) {
        //创建disruptor
        Disruptor<OrderEvent> disruptor = new Disruptor<>(
                new OrderEventFactory(),
                1024 * 1024,
                Executors.defaultThreadFactory(),
                ProducerType.SINGLE, //单生产者
                new YieldingWaitStrategy()  //等待策略
        );

        //设置消费者用于处理RingBuffer的事件
        disruptor.handleEventsWith(new OrderEventHandler());
        disruptor.start();

        //创建ringbuffer容器
        RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();
        //创建生产者
        OrderEventProducer eventProducer = new OrderEventProducer(ringBuffer);
        //发送消息
        for (int i = 0; i < 100; i++) {
            eventProducer.onData(i, "Hot " + i);
        }

        disruptor.shutdown();
    }

    // 1.事件
    @Data
    static public class OrderEvent {
        private long value;
        private String name;
    }

    // 2. 事件工厂
    static public class OrderEventFactory implements EventFactory<OrderEvent> {

        @Override
        public OrderEvent newInstance() {
            return new OrderEvent();
        }
    }

    // 3. 消息发布者
    static public class OrderEventProducer {
        //事件队列
        private RingBuffer<OrderEvent> ringBuffer;

        public OrderEventProducer(RingBuffer<OrderEvent> ringBuffer) {
            this.ringBuffer = ringBuffer;
        }

        public void onData(long value, String name) {
            // 获取事件队列 的下一个槽
            long sequence = ringBuffer.next();
            try {
                //获取消息（事件）
                OrderEvent orderEvent = ringBuffer.get(sequence);
                // 写入消息数据
                orderEvent.setValue(value);
                orderEvent.setName(name);
            } catch (Exception e) {
                // TODO  异常处理
                e.printStackTrace();
            } finally {
                System.out.println("生产者发送数据value:" + value + ",name:" + name);
                //发布事件
                ringBuffer.publish(sequence);
            }
        }
    }

    //4. 消息处理者（消费者）
    static public class OrderEventHandler implements EventHandler<OrderEvent> {

        @Override
        public void onEvent(OrderEvent event, long sequence, boolean endOfBatch) throws Exception {
            // TODO 消费逻辑
            System.out.println("消费者获取数据value:" + event.getValue() + ",name:" + event.getName());
        }
    }
}
